package com.duowan.realtime.scheduled.batch.processor;

import com.duowan.common.util.DateConvertUtils;
import com.duowan.common.util.Profiler;
import com.duowan.realtime.dao.redis.RedisDao;
import com.duowan.realtime.model.RedisKey;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * 某些kpi指标，redis里面存的是累计的值，需要将现在的值减去同样的key的昨天的值，生成一个新的kpi指标，方便之后存入mysql
 * 例如 20141023|sum_new_user --->2000 20141022|sum_new_user --->500
 * 则需要生成一个新的key   20141023|new_user --->1500
 *
 * @author tangsicheng
 * @version 1.0
 * @since 1.0
 */
public class RedisValueMinusYesterDayProcessor implements ItemProcessor<List<Map<String, String>>, List<Map<String, String>>> {


    private RedisDao redisDao;

    private static final Logger logger = LoggerFactory.getLogger(RedisValueMinusYesterDayProcessor.class);


    private String newKpiCode;

    private String getYesterdayPrefix(String prefix) {
        Date today = DateConvertUtils.parse(prefix, "yyyyMMdd");
        Date yesterday = DateUtils.addDays(today, -1);
        return DateFormatUtils.format(yesterday, "yyyyMMdd");

    }


    public void setNewKpiCode(String newKpiCode) {
        this.newKpiCode = newKpiCode;
    }


    private String getYesterDayKeyStr(String todayKeyStr) {
        RedisKey redisKey = RedisKey.factory(todayKeyStr);
        String prefix = redisKey.getPrefix();
        return redisKey.toString().replace(prefix, getYesterdayPrefix(prefix));
    }

    private String caculateNewValue(String todayValue, String yesterdayKpiValue) {
        // logger.info("today redis key " + todayValue + " redisYesterdayKey " + yesterdayKpiValue);
        if (StringUtils.isBlank(todayValue) && StringUtils.isBlank(yesterdayKpiValue)) {
            return "0";
        }
        String tempTodayValue = StringUtils.isNotBlank(todayValue) ? todayValue : "0";
        String tempYesterDayValue = StringUtils.isNotBlank(yesterdayKpiValue) ? yesterdayKpiValue : "0";
        try {
            int todayValueInt = Integer.parseInt(tempTodayValue);
            int yesterDayValueInt = Integer.parseInt(tempYesterDayValue);
            int newValue = todayValueInt - yesterDayValueInt;
            newValue = newValue < 0 ? 0 : newValue;
            return String.valueOf(newValue);
        } catch (NumberFormatException e) {
            logger.error("value 转换错误", e);
            return "0";
        }

    }

    @Override
    public List<Map<String, String>> process(List<Map<String, String>> item) throws Exception {
        Profiler.start();
        Profiler.enter("从历史机器和历史用户生成当天的新增的机器和新增用户");
        List<Map<String, String>> result = Lists.newLinkedList();
        for (Map<String, String> oneMap : item) {
            try {
                String redisKeyStr = oneMap.get("redisKey");
                String todayKpiValue = oneMap.get("redisValue");

                String redisYesterdayKey = getYesterDayKeyStr(redisKeyStr);

                logger.info("today redis key " + redisKeyStr + " redisYesterdayKey " + redisYesterdayKey);

                String yesterdayKpiValue = redisDao.getRedisTemplate().get(redisYesterdayKey);

                String newKpiValue = caculateNewValue(todayKpiValue, yesterdayKpiValue);
                String newRedisKey = getNewKey(redisKeyStr);

                Map<String, String> tempMap = ImmutableMap.of("redisKey", newRedisKey, "redisValue", newKpiValue);
                result.add(tempMap);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                continue;
            }
        }
        result.addAll(item);
        Profiler.release();
        logger.info(Profiler.dump());
        return result;
    }

    /**
     * 更具新的kpiCode生成一个新的key
     */
    private String getNewKey(String redisKeyStr) {
        RedisKey redisKey = RedisKey.factory(redisKeyStr);
        return RedisKey.serRedisKey(redisKey.getPrefix(), newKpiCode, redisKey.getOrderKVMap());
    }

    public void setRedisDao(RedisDao redisDao) {
        this.redisDao = redisDao;
    }
}
